/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
miniob is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
         http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

//
// Created by Wangyunlai on 2022/6/6.
//

#include "sql/stmt/select_stmt.h"
#include "sql/stmt/filter_stmt.h"
#include "sql/stmt/join_stmt.h"
#include "common/log/log.h"
#include "common/lang/string.h"
#include "storage/db/db.h"
#include "storage/table/table.h"

SelectStmt::~SelectStmt()
{
  if (nullptr != filter_stmt_)
  {
    delete filter_stmt_;
    filter_stmt_ = nullptr;
  }
  if(nullptr!=join_stmt_){
    delete join_stmt_;
    join_stmt_=nullptr;
  }
}
SelectStmt::SelectStmt(const SelectStmt &other)
{
  this->query_fields_ = other.query_fields_;
  this->tables_ = other.tables_;
  if (other.filter_stmt_ != nullptr)
  {
    this->filter_stmt_ = new FilterStmt(*other.filter_stmt_);
  }
  if (other.join_stmt_ != nullptr)
  {
    this->join_stmt_ = new JoinStmt(*other.join_stmt_);
  }
  this->func_types_ = other.func_types_;
  this->func_fields_ = other.func_fields_;
  this->order_fields_ = other.order_fields_;
  this->order_types_ = other.order_types_;
  this->cols_alias_=other.cols_alias_;
}
SelectStmt &SelectStmt::operator=(const SelectStmt &other)
{
  this->query_fields_ = other.query_fields_;
  this->tables_ = other.tables_;
  if (other.filter_stmt_ != nullptr)
  {
    this->filter_stmt_ = new FilterStmt(*other.filter_stmt_);
  }
  if (other.join_stmt_ != nullptr)
  {
    this->join_stmt_ = new JoinStmt(*other.join_stmt_);
  }
  this->func_types_ = other.func_types_;
  this->func_fields_ = other.func_fields_;
  this->order_types_ = other.order_types_;
  this->order_fields_ = other.order_fields_;
  this->cols_alias_=other.cols_alias_;
  return *this;
}
static void wildcard_fields(Table *table, std::string table_alias,std::vector<Field> &field_metas,std::vector<std::string>& cols_alias,bool with_table_name)
{
  const TableMeta &table_meta = table->table_meta();
  const int field_num = table_meta.field_num();
  for (int i = table_meta.sys_field_num(); i < field_num; i++)
  {
    Field field=Field(table, table_meta.field(i));
    std::string col_alias=field.field_name();
    if(with_table_name){
      col_alias=table_alias+"."+col_alias;
    }
    cols_alias.push_back(col_alias);
    field_metas.push_back(field);
  }
}

RC SelectStmt::create(Db *db, const SelectSqlNode &select_sql, Stmt *&stmt, std::vector<std::string> p_tables,
std::vector<std::string> p_alias)
{
  if (nullptr == db)
  {
    LOG_WARN("invalid argument. db is null");
    return RC::INVALID_ARGUMENT;
  }
  // collect tables in `from` statement
 std::vector<std::string> tables_alias;
  std::vector<std::string> cols_alias;
  std::vector<Table *> tables;
  std::unordered_map<std::string, Table *> table_map;
  for (size_t i = 0; i < select_sql.joins.size(); i++)
  {
    const char *table_name = select_sql.joins[i].table_name.c_str();
    const char *table_alias=select_sql.joins[i].table_alias.c_str();
    if (nullptr == table_name)
    {
      LOG_WARN("invalid argument. relation name is null. index=%d", i);
      return RC::INVALID_ARGUMENT;
    }

    Table *table = db->find_table(table_name);
    if (nullptr == table)
    {
      LOG_WARN("no such table. db=%s, table_name=%s", db->name(), table_name);
      return RC::SCHEMA_TABLE_NOT_EXIST;
    }

      tables.push_back(table);
      tables_alias.push_back(table_alias);
      table_map.insert(std::pair<std::string, Table *>(table_alias, table));
  }
  // collect query fields in `select` statement
  std::vector<Field> query_fields;
  std::vector<RelAttrSqlNode> attributes;
  std::vector<FuncSqlNode> funcs;
  std::vector<FuncType> func_types;
  std::vector<std::vector<Field>> func_fields;
  std::vector<Field> order_fields;
  std::vector<OrderType> order_types;
  std::vector<OrderSqlNode> order_nodes = select_sql.orders;
  funcs = select_sql.funcs;
  attributes = select_sql.attributes;
  int table_num=tables.size();
  bool with_table_name=(table_num>1);
  if (funcs.size() == 0 && attributes.size() == 0)
  {
    LOG_WARN("invalid argument null attribute_name");
    return RC::INVALID_ARGUMENT;
  }
  if (funcs.size() == 0)
  {
    for (int i = 0; i < attributes.size(); i++)
    {
      const RelAttrSqlNode &relation_attr = attributes[i];
      if (common::is_blank(relation_attr.attribute_name.c_str()))
      {
        LOG_WARN("invalid argument null attribute_name and attr");
        return RC::INVALID_ARGUMENT;
      }
      if (common::is_blank(relation_attr.relation_name.c_str()) &&
          0 == strcmp(relation_attr.attribute_name.c_str(), "*"))
      {
        if (attributes.size() > 1)
        {
          LOG_WARN("invalid argument mixed * and attr");
          return RC::INVALID_ARGUMENT;
        }
        for (int i=0;i<table_num;i++)
        {
          Table* table=tables[i];
          std::string table_alias=tables_alias[i];
          wildcard_fields(table, table_alias,query_fields,cols_alias,with_table_name);
        }
      }
      else if (!common::is_blank(relation_attr.relation_name.c_str()))
      {
        const char *table_name = relation_attr.relation_name.c_str();
        const char *field_name = relation_attr.attribute_name.c_str();
        const char *field_alias=relation_attr.attribute_alias.c_str();
        bool has_alias=relation_attr.has_alias;
        
        if (0 == strcmp(table_name, "*"))
        {
          if (0 != strcmp(field_name, "*"))
          {
            LOG_WARN("invalid field name while table is *. attr=%s", field_name);
            return RC::SCHEMA_FIELD_MISSING;
          }
          if (attributes.size() > 1)
          {
            LOG_WARN("invalid argument mixed * and attr");
            return RC::INVALID_ARGUMENT;
          }
          
        for (int i=0;i<table_num;i++)
        {
          Table* table=tables[i];
          std::string table_alias=tables_alias[i];
          wildcard_fields(table, table_alias,query_fields,cols_alias,with_table_name);
        }
        }
        else
        {
          auto iter = table_map.find(table_name);
          if (iter == table_map.end())
          {
            LOG_WARN("no such table in from list: %s", table_name);
            return RC::SCHEMA_FIELD_MISSING;
          }

          Table *table = iter->second;
          if (0 == strcmp(field_name, "*"))
          {
            if (attributes.size() > 1)
            {
              LOG_WARN("invalid argument mixed * and attr");
              return RC::INVALID_ARGUMENT;
            }
            wildcard_fields(table,table_name, query_fields,cols_alias,with_table_name);
          }
          else
          {
            const FieldMeta *field_meta = table->table_meta().field(field_name);
            if (nullptr == field_meta)
            {
              LOG_WARN("no such field. field=%s.%s.%s", db->name(), table->name(), field_name);
              return RC::SCHEMA_FIELD_MISSING;
            }
            if(has_alias){
              cols_alias.push_back(field_alias);
            }
            else {
              if(with_table_name){
                std::string alias=table_name;
                alias=alias+"."+field_name;
                cols_alias.push_back(alias);
              }
            }
            query_fields.push_back(Field(table, field_meta));
          }
        }
      }
      else
      {
        if (tables.size() != 1)
        {
          LOG_WARN("invalid. I do not know the attr's table. attr=%s", relation_attr.attribute_name.c_str());
          return RC::SCHEMA_FIELD_MISSING;
        }

        Table *table = tables[0];
        const FieldMeta *field_meta = table->table_meta().field(relation_attr.attribute_name.c_str());
        if (nullptr == field_meta)
        {
          LOG_WARN("no such field. field=%s.%s.%s", db->name(), table->name(), relation_attr.attribute_name.c_str());
          return RC::SCHEMA_FIELD_MISSING;
        }
        cols_alias.push_back(relation_attr.attribute_alias);
        query_fields.push_back(Field(table, field_meta));
      }
    }
  }
  else
  {
    if (attributes.size() > 0)
    {
      LOG_WARN("invalid argument mixed funs and attributes");
      return RC::INVALID_ARGUMENT;
    }
    for (FuncSqlNode node : funcs)
    {
      std::vector<RelAttrSqlNode> attrs = node.func_attributes;
      if (attrs.size() == 0)
      {
        LOG_WARN("invalid argument none func attributes");
        return RC::INVALID_ARGUMENT;
      }
      std::vector<Field> tmp_fields;
      for (RelAttrSqlNode attr : attrs)
      {
        if (common::is_blank(attr.attribute_name.c_str()))
        {
          LOG_WARN("invalid argument null attribute_name and attr");
          return RC::INVALID_ARGUMENT;
        }
        if (common::is_blank(attr.relation_name.c_str()) &&
            0 == strcmp(attr.attribute_name.c_str(), "*"))
        {
          if (attrs.size() > 1)
          {
            LOG_WARN("invalid argument mixed * and attr");
            return RC::INVALID_ARGUMENT;
          }
          for (int i=0;i<table_num;i++)
        {
          Table* table=tables[i];
          std::string table_alias=tables_alias[i];
          wildcard_fields(table, table_alias,tmp_fields,cols_alias,with_table_name);
        }
        }
        else if (!common::is_blank(attr.relation_name.c_str()))
        {
          const char *table_name = attr.relation_name.c_str();
          const char *field_name = attr.attribute_name.c_str();
          if (0 == strcmp(table_name, "*"))
          {
            if (0 != strcmp(field_name, "*"))
            {
              LOG_WARN("invalid field name while table is *. attr=%s", field_name);
              return RC::SCHEMA_FIELD_MISSING;
            }
            if (attrs.size() > 1)
            {
              LOG_WARN("invalid argument mixed * and attr");
              return RC::INVALID_ARGUMENT;
            }
            for (int i=0;i<table_num;i++)
        {
          Table* table=tables[i];
          std::string table_alias=tables_alias[i];
          wildcard_fields(table, table_alias,tmp_fields,cols_alias,with_table_name);
        }
          }
          else
          {
            auto iter = table_map.find(table_name);
            if (iter == table_map.end())
            {
              LOG_WARN("no such table in from list: %s", table_name);
              return RC::SCHEMA_FIELD_MISSING;
            }

            Table *table = iter->second;
            if (0 == strcmp(field_name, "*"))
            {
              if (attributes.size() > 1)
              {
                LOG_WARN("invalid argument mixed * and attr");
                return RC::INVALID_ARGUMENT;
              }
              wildcard_fields(table,table_name,tmp_fields,cols_alias,with_table_name);
            }
            else
            {
              const FieldMeta *field_meta = table->table_meta().field(field_name);
              if (nullptr == field_meta)
              {
                LOG_WARN("no such field. field=%s.%s.%s", db->name(), table->name(), field_name);
                return RC::SCHEMA_FIELD_MISSING;
              }
              cols_alias.push_back(attr.attribute_alias);
              tmp_fields.push_back(Field(table, field_meta));
            }
          }
        }
        else
        {
          if (tables.size() != 1)
          {
            LOG_WARN("invalid. I do not know the attr's table. attr=%s", attr.attribute_name.c_str());
            return RC::SCHEMA_FIELD_MISSING;
          }

          Table *table = tables[0];
          const FieldMeta *field_meta = table->table_meta().field(attr.attribute_name.c_str());
          if (nullptr == field_meta)
          {
            LOG_WARN("no such field. field=%s.%s.%s", db->name(), table->name(), attr.attribute_name.c_str());
            return RC::SCHEMA_FIELD_MISSING;
          }

          tmp_fields.push_back(Field(table, field_meta));
        }
      }
      if (node.func_type != FuncType::F_COUNT && tmp_fields.size() > 1)
      {
        LOG_WARN("invalid argument multy func attributes");
        return RC::INVALID_ARGUMENT;
      }
      if (node.func_type == FuncType::F_AVG || node.func_type == FuncType::F_SUM)
      {
        Field field = tmp_fields[0];
        if (field.attr_type() != AttrType::INTS && field.attr_type() != AttrType::FLOATS)
        {
          LOG_WARN("invalid argument wrong func attribute type");
          return RC::INVALID_ARGUMENT;
        }
      }
      query_fields.push_back(tmp_fields[0]);
      func_fields.push_back(tmp_fields);
      func_types.push_back(node.func_type);
    }
  }
  LOG_INFO("got %d tables in from stmt and %d fields in query stmt", tables.size(), query_fields.size());

  Table *default_table = nullptr;
  if (tables.size() == 1)
  {
    default_table = tables[0];
  }

  JoinStmt *join_stmt = nullptr;
  RC rc = JoinStmt::create(db, select_sql.joins.data(), static_cast<int>(select_sql.joins.size()), join_stmt);
  if (rc != RC::SUCCESS)
  {
    LOG_WARN("cannot construct join stmt");
    return rc;
  }
  // create filter statement in `where` statement
  FilterStmt *filter_stmt = nullptr;
  rc = FilterStmt::create(db,
                          default_table,
                          query_fields,
                          &table_map,
                          select_sql.conditions.data(),
                          static_cast<int>(select_sql.conditions.size()),
                          filter_stmt, p_tables);
  if (rc != RC::SUCCESS)
  {
    LOG_WARN("cannot construct filter stmt");
    return rc;
  }
  if (order_nodes.size() > 0)
  {
    for (int i = 0; i < order_nodes.size(); i++)
    {
      const RelAttrSqlNode &relation_attr = order_nodes[i].order_attribute;
      const OrderType &order_type = order_nodes[i].order_type;
      if (common::is_blank(relation_attr.attribute_name.c_str()))
      {
        LOG_WARN("invalid argument null attribute_name and attr");
        return RC::INVALID_ARGUMENT;
      }
      if (0 == strcmp(relation_attr.attribute_name.c_str(), "*"))
      {
        LOG_WARN("invalid * in order attr");
        return RC::INVALID_ARGUMENT;
      }
      if (common::is_blank(relation_attr.relation_name.c_str()))
      {

        if (tables.size() > 1)
        {
          LOG_WARN("invalid blank table_name in order attr when multy_table order");
          return RC::INVALID_ARGUMENT;
        }
        Table *table = tables[0];
        const char *field_name = relation_attr.attribute_name.c_str();
        const FieldMeta *field_meta = table->table_meta().field(field_name);
        if (nullptr == field_meta)
        {
          LOG_WARN("no such field. field=%s.%s.%s", db->name(), table->name(), field_name);
          return RC::SCHEMA_FIELD_MISSING;
        }
        order_fields.push_back(Field(table, field_meta));
        order_types.push_back(order_type);
      }
      else
      {
        const char *table_name = relation_attr.relation_name.c_str();
        const char *field_name = relation_attr.attribute_name.c_str();
        if (0 == strcmp(table_name, "*"))
        {
          LOG_WARN("invalid table * in order attr");
          return RC::INVALID_ARGUMENT;
        }
        else
        {
          auto iter = table_map.find(table_name);
          if (iter == table_map.end())
          {
            LOG_WARN("no such table in from list: %s", table_name);
            return RC::SCHEMA_FIELD_MISSING;
          }
          Table *table = iter->second;
          const FieldMeta *field_meta = table->table_meta().field(field_name);
          if (nullptr == field_meta)
          {
            LOG_WARN("no such field. field=%s.%s.%s", db->name(), table->name(), field_name);
            return RC::SCHEMA_FIELD_MISSING;
          }
          order_fields.push_back(Field(table, field_meta));
          order_types.push_back(order_type);
        }
      }
    }
  }

  // everything alright
  SelectStmt *select_stmt = new SelectStmt();
  // TODO add expression copy
  select_stmt->tables_.swap(tables);
  select_stmt->query_fields_.swap(query_fields);
  select_stmt->filter_stmt_ = filter_stmt;
  select_stmt->join_stmt_ = join_stmt;
  select_stmt->func_types_.swap(func_types);
  select_stmt->func_fields_.swap(func_fields);
  select_stmt->order_types_.swap(order_types);
  select_stmt->order_fields_.swap(order_fields);
  stmt = select_stmt;
  return RC::SUCCESS;
}
